package TFIDF;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.db.DBConfiguration;
import org.apache.hadoop.mapreduce.lib.db.DBInputFormat;
import org.apache.hadoop.mapreduce.lib.db.DBOutputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.log4j.BasicConfigurator;

import java.net.URI;

public class DriTest {
    public static void main(String[] args) throws Exception {
        BasicConfigurator.configure();
        Configuration conf = new Configuration();
        DBConfiguration.configureDB(conf, "com.mysql.jdbc.Driver", "jdbc:mysql://localhost:3306/data?useUnicode=true&characterEncoding=UTF-8", "root", "200028");
        Job job = Job.getInstance(conf);
        job.setJarByClass(DriTest.class);

        job.setMapperClass(MapTest01.class);
        job.setReducerClass(RedTest01.class);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(DoubleWritable.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(DoubleWritable.class);

        job.setPartitionerClass(MyPartition.class);
        job.setNumReduceTasks(2);

        String[] f1 = {"rank", "name", "actor", "grade", "num", "common"};
        DBInputFormat.setInput(job, receiveData.class, "douban", null, null, f1);
        FileOutputFormat.setOutputPath(job, new Path("D:\\MP\\TFIDF\\output1"));
        boolean b1 = job.waitForCompletion(true);
        if (b1) {
            Configuration conf2 = new Configuration();
            Job job2 = Job.getInstance(conf2);
            job2.setJarByClass(DriTest.class);

            job2.setMapperClass(MapTest02.class);
            job2.setReducerClass(RedTest02.class);

            job2.setMapOutputKeyClass(Text.class);
            job2.setMapOutputValueClass(IntWritable.class);

            job2.setOutputKeyClass(Text.class);
            job2.setOutputValueClass(IntWritable.class);

            job2.setGroupingComparatorClass(MyGroup.class);

            FileInputFormat.setInputPaths(job2, "D:\\MP\\TFIDF\\output1\\part-r-00001");
            FileOutputFormat.setOutputPath(job2, new Path("D:\\MP\\TFIDF\\output2"));
            boolean b2 = job2.waitForCompletion(true);
            if (b2) {
                Configuration conf3 = new Configuration();
                DBConfiguration.configureDB(conf3, "com.mysql.jdbc.Driver", "jdbc:mysql://localhost:3306/data?useUnicode=true&characterEncoding=UTF-8", "root", "200028");
                Job job3 = Job.getInstance(conf3);
                job3.setJarByClass(DriTest.class);

                job3.setMapperClass(MapTest03.class);
                job3.setReducerClass(RedTest03.class);

                job3.setMapOutputKeyClass(NullWritable.class);
                job3.setMapOutputValueClass(ResultData.class);

                job3.setOutputKeyClass(ResultData.class);
                job3.setOutputValueClass(NullWritable.class);

                URI uri[] = new URI[2];
                uri[0] = new URI("file:///D:/MP/TFIDF/output1/part-r-00000");
                uri[1] = new URI("file:///D:/MP/TFIDF/output1/part-r-00001");
                job3.setCacheFiles(uri);
                FileInputFormat.setInputPaths(job3, "D:\\MP\\TFIDF\\output2\\part-r-00000");
                String[] f2 = {"FilmName", "word", "TF", "IDF", "TF_IDF"};
                DBOutputFormat.setOutput(job3, "resultData", f2);
//                FileOutputFormat.setOutputPath(job3, new Path("D:\\MP\\TFIDF\\output3"));
                System.exit(job3.waitForCompletion(true) ? 0 : 1);
            }
        }
    }
}
